package com.deep.cdc;

import com.deep.bean.ConnectConfig;
import com.deep.bean.TransformConfig;
import com.deep.common.ArgsConfig;
import com.deep.common.ArgsUtil;
import com.deep.common.ProfileConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SyncTranInst {
    public static void main(String[] args) {
        ArgsConfig argsConfig = ArgsUtil.getMap(args);
        Integer recid = argsConfig.getRecid();

        TransformConfig transformConfig = ProfileConfig.getTransformConfig(argsConfig.getProfile());
        ConnectConfig sourceConfig = transformConfig.getSource();
        ConnectConfig targetConfig = transformConfig.getTarget();


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建用于输出的表
        String createInputDDL = "create table source_t_wf_transinst " +
                "( " +
                "ACTID  int , " +
                "ITEMID  int , " +
                "TRANSID  int , " +
                "RECID  int , " +
                "WORKITEMID  int , " +
                "ACTDEFID  int , " +
                "NEXTACTID  int , " +
                "NEXTACTDEFID  int , " +
                "NEXTPARTID  int , " +
                "NEXTROLEPARTID  int , " +
                "TRANSSTATEID  int , " +
                "    PRIMARY KEY (`TRANSID`) NOT ENFORCED  " +
                ")WITH (  " +
                "    'connector' = 'mysql-cdc', " +
                "    'hostname' = '" + sourceConfig.getHostname() + "',  " +
                "    'port' = '" + sourceConfig.getPort() + "',   " +
                "    'username' = '" + sourceConfig.getUsername() + "',  " +
                "    'password' = '" + sourceConfig.getPassword() + "',   " +
                "    'database-name' = '" + sourceConfig.getDatabaseName() + "',  " +
                "    'table-name' = 't_wf_transinst', " +
                "   'debezium.skipped.operations'='d'  " +
                ") ";
        tableEnv.executeSql(createInputDDL);


        //创建用于输出的表
        String createOutputDDL = "create table target_t_wf_transinst  " +
                "( " +
                "ACTID  int , " +
                "ITEMID  int , " +
                "TRANSID  int , " +
                "RECID  int , " +
                "WORKITEMID  int , " +
                "ACTDEFID  int , " +
                "NEXTACTID  int , " +
                "NEXTACTDEFID  int , " +
                "NEXTPARTID  int , " +
                "NEXTROLEPARTID  int , " +
                "TRANSSTATEID  int , " +
                "    PRIMARY KEY (`TRANSID`) NOT ENFORCED  " +
                ")WITH (   " +
                " 'connector' = 'jdbc',  " +
                " 'driver' = 'com.mysql.cj.jdbc.Driver',  " +
                " 'url' = 'jdbc:mysql://"+ targetConfig.getHostname() + ":" + targetConfig.getPort() + "/" + targetConfig.getDatabaseName() + "?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC',  " +
                " 'username' = '" + targetConfig.getUsername() + "',  " +
                "  'password' = '" + targetConfig.getPassword() + "',   " +
                "  'table-name' = 't_wf_transinst',  " +
                "  'connection.max-retry-timeout' = '60s'   " +
                ")";
        tableEnv.executeSql(createOutputDDL);


        TableResult tableResult = tableEnv.executeSql("insert into target_t_wf_transinst select * from source_t_wf_transinst where RECID >  " + recid);

        tableResult.print();

    }
}
